package com.haomin.searcher.indexer.core;

import com.haomin.searcher.indexer.aop.Timing;
import com.haomin.searcher.indexer.mapper.IndexDatabaseMapper;
import com.haomin.searcher.indexer.model.Document;
import com.haomin.searcher.indexer.model.InvertedRecord;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

/**
 * @author haomin
 * @date 2022/10/05 10:38
 **/

@Slf4j
@Service
public class IndexManager {
    private final IndexDatabaseMapper mapper;
    private final ExecutorService executorService;

    @Autowired
    public IndexManager(IndexDatabaseMapper mapper, ExecutorService executorService) {
        this.mapper = mapper;
        this.executorService = executorService;
    }

    // 先批量生成、保存正排索引（单线程版本）
    public void saveForwardIndexes(List<Document> documentList) {
        // 1. 批量插入时，每次插入多少条记录（由于每条记录比较大，所以这里使用 10 条就够了）
        int batchSize = 10;
        // 2. 一共需要执行多少次 SQL？   向上取整(documentList.size() / batchSize)
        int listSize = documentList.size();
        int times = (int) Math.ceil(1.0 * listSize / batchSize);    // ceil(天花板): 向上取整
        log.debug("一共需要 {} 批任务。", times);

        // 3. 开始分批次插入
        for (int i = 0; i < listSize; i += batchSize) {
            // 从 documentList 中截取这批要插入的 文档列表（使用 List.subList(int from, int to)
            int from = i;
            int to = Integer.min(from + batchSize, listSize);

            List<Document> subList = documentList.subList(from, to);

            // 针对这个 subList 做批量插入
            mapper.batchInsertForwardIndexes(subList);
        }
    }
    @Timing("构建 + 保存正排索引 —— 多线程版本")
    @SneakyThrows
    public void saveForwardIndexesConcurrent(List<Document> documentList) {
        // 1. 批量插入时，每次插入多少条记录（由于每条记录比较大，所以这里使用 10 条就够了）
        int batchSize = 10;
        // 2. 一共需要执行多少次 SQL？   向上取整(documentList.size() / batchSize)
        int listSize = documentList.size();
        int times = (int) Math.ceil(1.0 * listSize / batchSize);    // ceil(天花板): 向上取整
        log.debug("一共需要 {} 批任务。", times);

        CountDownLatch latch = new CountDownLatch(times);   // 统计每个线程的完全情况，初始值是 times(一共多少批)

        // 3. 开始分批次插入
        for (int i = 0; i < listSize; i += batchSize) {
            // 从 documentList 中截取这批要插入的 文档列表（使用 List.subList(int from, int to)
            int from = i;
            int to = Integer.min(from + batchSize, listSize);

            Runnable task = () -> { // 内部类 / lambda 表达式里如果用到了外部变量，外部变量必须的 final（或者隐式 final 的变量）
                List<Document> subList = documentList.subList(from, to);

                // 针对这个 subList 做批量插入
                mapper.batchInsertForwardIndexes(subList);

                latch.countDown();      //  每次任务完成之后，countDown()，让 latch 的个数减一
            };

            executorService.submit(task);   // 主线程只负责把一批批的任务提交到线程池，具体的插入工作，由线程池中的线程完成
        }

        // 4. 循环结束，只意味着主线程把任务提交完成了，但任务有没有做完是不知道的
        // 主线程等在 latch 上，只到 latch 的个数变成 0，也就是所有任务都已经执行完了
        latch.await();
    }


    // 先批量生成、保存倒排索引（单线程版本）
    @SneakyThrows
    public void saveInvertedIndexes(List<Document> documentList) {
        int batchSize = 10000;  // 批量插入时，最多 10000 条

        List<InvertedRecord> recordList = new ArrayList<>();    // 放这批要插入的数据

        for (Document document : documentList) {
            Map<String, Integer> wordToWeight = document.segWordAndCalcWeight();
            for (Map.Entry<String, Integer> entry : wordToWeight.entrySet()) {
                String word = entry.getKey();
                int docId = document.getDocId();
                int weight = entry.getValue();

                InvertedRecord record = new InvertedRecord(word, docId, weight);

                recordList.add(record);

                // 如果 recordList.size() == batchSize，说明够一次插入了
                if (recordList.size() == batchSize) {
                    mapper.batchInsertInvertedIndexes(recordList);  // 批量插入
                    recordList.clear();                             // 清空 list，视为让 list.size() = 0
                }
            }
        }

        // recordList 还剩一些，之前放进来，但还不够 batchSize 个的，所以最后再批量插入一次
        mapper.batchInsertInvertedIndexes(recordList);  // 批量插入
        recordList.clear();
    }


    @Timing("构建 + 保存倒排索引 —— 多线程版本")
    @SneakyThrows
    public void saveInvertedIndexesConcurrent(List<Document> documentList) {
        int batchSize = 10000;  // 批量插入时，最多 10000 条
        int groupSize = 50;
        int listSize = documentList.size();
        int times = (int) Math.ceil(listSize * 1.0 / groupSize);
        CountDownLatch latch = new CountDownLatch(times);

        for (int i = 0; i < listSize; i += groupSize) {
            int from = i;
            int to = Integer.min(from + groupSize, listSize);
            List<Document> subList = documentList.subList(from, to);
            Runnable task = new InvertedInsertTask(latch, batchSize, subList, mapper);
            executorService.submit(task);
        }

        latch.await();
    }
    // 多线程处理分批次构建倒排索引任务
    static class InvertedInsertTask implements Runnable {
        private final CountDownLatch latch;
        private final int batchSize;
        private final List<Document> documentList;
        private final IndexDatabaseMapper mapper;

        InvertedInsertTask(CountDownLatch latch, int batchSize, List<Document> documentList, IndexDatabaseMapper mapper) {
            this.latch = latch;
            this.batchSize = batchSize;
            this.documentList = documentList;
            this.mapper = mapper;
        }

        @Override
        public void run() {
            List<InvertedRecord> recordList = new ArrayList<>();    // 放这批要插入的数据

            for (Document document : documentList) {
                Map<String, Integer> wordToWeight = document.segWordAndCalcWeight();
                for (Map.Entry<String, Integer> entry : wordToWeight.entrySet()) {
                    String word = entry.getKey();
                    int docId = document.getDocId();
                    int weight = entry.getValue();

                    InvertedRecord record = new InvertedRecord(word, docId, weight);

                    recordList.add(record);

                    // 如果 recordList.size() == batchSize，说明够一次插入了
                    if (recordList.size() == batchSize) {
                        mapper.batchInsertInvertedIndexes(recordList);  // 批量插入
                        recordList.clear();                             // 清空 list，视为让 list.size() = 0
                    }
                }
            }

            // recordList 还剩一些，之前放进来，但还不够 batchSize 个的，所以最后再批量插入一次
            mapper.batchInsertInvertedIndexes(recordList);  // 批量插入
            recordList.clear();

            latch.countDown();
        }
    }
}



// 原版本（一个一个插入构建索引）

//@Slf4j
//@Service
//public class IndexManager {
//    private final List<IndexDocument> forwardIndex = new ArrayList<>();
//    private final Map<String, List<DocWeight>> invertedIndex = new HashMap<>();
//
//    // 持久化保存
//    private final ObjectMapper objectMapper;
//    private final IndexerProperties properties;
//    // 保存到数据库
//    private final DatabaseMapper databaseMapper;
//
//    public IndexManager(ObjectMapper objectMapper, IndexerProperties properties, DatabaseMapper databaseMapper) {
//        this.objectMapper = objectMapper;
//        this.properties = properties;
//        this.databaseMapper = databaseMapper;
//    }
//
//    public void add(Document document) {
//        // 添加正排索引
//        int docId = forwardIndex.size();
//        IndexDocument indexDocument = new IndexDocument(docId, document);
//        forwardIndex.add(indexDocument);    // 下标(docId) -> IndexDocument
//
//        // 添加倒排索引
//        buildInvertedIndex(indexDocument);
//    }
//
//    public void printForwardIndex() {
//        int size = forwardIndex.size();
//        for (int i = 0; i < size; i++) {
//            int docId = i;
//            IndexDocument indexDocument = forwardIndex.get(docId);
//            System.out.printf("docId: %d -> IndexDocument: %s\n", docId, indexDocument);
//        }
//    }
//
//    public void printInvertedIndex() {
//        Set<String> wordSet = invertedIndex.keySet();
//        for (String word : wordSet) {
//            System.out.printf("%s: \n", word);
//            List<DocWeight> docWeightList = invertedIndex.get(word);
//            for (DocWeight docWeight : docWeightList) {
//                System.out.printf("    docWeight: %s\n", docWeight);
//            }
//        }
//    }
//
//    private void buildInvertedIndex(IndexDocument indexDocument) {
//        int docId = indexDocument.getDocId();
//        Document document = indexDocument.getDocument();
//
//        // 1 首先对标题做分词
//        List<String> wordListInTitle = ToAnalysis.parse(document.getTitle())
//                .getTerms()
//                .stream()
//                .map(Term::getName)
//                .collect(Collectors.toList());
//
//        // 2 其次对正文做分词
//        List<String> wordListInContent = ToAnalysis.parse(document.getContent())
//                .getTerms()
//                .stream()
//                .map(Term::getName)
//                .collect(Collectors.toList());
//
//        // 3 根据每个次，计算该文档的每个词的出现次数（分别是标题和正文中的）
//        Map<String, Integer> wordInTitleMap = new HashMap<>();
//        for (String word : wordListInTitle) {
//            int count = wordInTitleMap.getOrDefault(word, 0);
//            count++;
//            wordInTitleMap.put(word, count);
//        }
//
//        Map<String, Integer> wordInContentMap = new HashMap<>();
//        for (String word : wordListInContent) {
//            int count = wordInContentMap.getOrDefault(word, 0);
//            count++;
//            wordInContentMap.put(word, count);
//        }
//
//        // 4 针对每个单词，做倒排索引创建
//        Set<String> allWordSet = new HashSet<>(wordInTitleMap.keySet());
//        allWordSet.addAll(wordInContentMap.keySet());
//
//        for (String word : allWordSet) {
//            // 计算每个单词的权重
//            DocWeight docWeight = new DocWeight(docId, word, wordInTitleMap, wordInContentMap);
//
//            List<DocWeight> docWeightList = invertedIndex.get(word);
//            // 如果这个单词之前没有出现过，则 docWeightList == null
//            // 这种情况下，我们需要特殊处理
//            if (docWeightList == null) {
//                docWeightList = new ArrayList<>();
//                invertedIndex.put(word, docWeightList);
//            }
//
//            // 如何什么情况下，将计算出的 docId + weight 放入 invertedIndex  word -> [ {docId, weight}, ]
//            docWeightList.add(docWeight);
//        }
//    }
//
//    // 保存成文件
//    @SneakyThrows
//    public void save1() {
//        // 使用 JSON 将两个索引序列后出来，然后分别保存成两个文件
//        // 确保 properties.getIndexRootPath() 是个目录 && 一定存在
//
//        // forward.json
//        File forwardFile = new File(properties.getIndexRootPath(), "forward.json");
//        objectMapper.writeValue(forwardFile, forwardIndex);
//        log.debug("正排索引已保存。{}", forwardFile.getCanonicalPath());
//
//        // inverted.json
//        File invertedFile = new File(properties.getIndexRootPath(), "inverted.json");
//        objectMapper.writeValue(invertedFile, invertedIndex);
//        log.debug("倒排索引已保存。{}", invertedFile.getCanonicalPath());
//    }
//
//    // 保存成数据库
//    public void save() {
//        // 先做正排索引的插入（一条条地插入，其实性能不好，这里暂时不管了）
//        log.debug("将索引保存到数据库中。");
//        for (IndexDocument indexDocument : forwardIndex) {
//            databaseMapper.insertDocument(
//                    indexDocument.getDocId(),
//                    indexDocument.getDocument().getTitle(),
//                    indexDocument.getDocument().getUrl(),
//                    indexDocument.getDocument().getContent()
//            );
//        }
//        log.debug("正排索引保存成功。");
//
//        // 再做倒排索引的插入
//        Set<Map.Entry<String, List<DocWeight>>> entries = invertedIndex.entrySet();
//        for (Map.Entry<String, List<DocWeight>> entry : entries) {
//            String word = entry.getKey();
//            List<DocWeight> weightList = entry.getValue();
//            // 需要先做 weightList -> weights 表 的插入，根据插入结果，得到 wid，在做 word_to_weights 表的插入
//            for (DocWeight docWeight : weightList) {
//                WeightDO weightDO = new WeightDO(docWeight);
//
//                databaseMapper.insertWeight(weightDO);
//
//                databaseMapper.insertWordToWeight(word, weightDO.getWid());
//            }
//        }
//        log.debug("倒排索引保存成功。");
//    }
//}